package net.sina.realtime.traffic.job3;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.SneakyThrows;
import net.sina.realtime.traffic.bean.MonitorInfo;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Properties;

public class TaoPaiCarController {
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Violation {
        private int id;
        private String car;
        private String violation;
        private Long createTime;
    }

    public static void main(String[] args) throws Exception {
        // 1. 准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(1); // 简化调试

        // 2. 加载数据 - 读取Kafka
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "node101:9092,node102:9092");
        properties.setProperty("group.id", "g1");
        FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>("topic-car", new SimpleStringSchema(), properties);
        kafkaSource.setStartFromEarliest(); // 从最早的消息开始消费
        DataStreamSource<String> dataStreamSource = env.addSource(kafkaSource);
        dataStreamSource.print("原始数据: ");

        // 3. 数据处理转换
        // 3.1 转换为MonitorInfo对象
        DataStream<MonitorInfo> ds1 = dataStreamSource.map(new MapFunction<String, MonitorInfo>() {
            @Override
            public MonitorInfo map(String jsonStr) throws Exception {
                return JSON.parseObject(jsonStr, MonitorInfo.class);
            }
        });

        // 3.2 添加水位线和事件时间
        SingleOutputStreamOperator<MonitorInfo> mapStream = ds1.assignTimestampsAndWatermarks(
                WatermarkStrategy.<MonitorInfo>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<MonitorInfo>() {
                            @SneakyThrows
                            @Override
                            public long extractTimestamp(MonitorInfo element, long recordTimestamp) {
                                // 假设element.getActionTime()是秒级时间戳，转换为毫秒
                                return element.getActionTime() * 1000L;
                            }
                        })
        );

//        // 3.3 套牌检测逻辑
        SingleOutputStreamOperator<Violation> resultDs = mapStream
                .keyBy(MonitorInfo::getCar)
                .flatMap(new RichFlatMapFunction<MonitorInfo, Violation>() {

                    private transient ValueState<MonitorInfo> previousInfoState;

                    @Override
                    public void open(Configuration parameters) {
                        ValueStateDescriptor<MonitorInfo> stateDescriptor =
                                new ValueStateDescriptor<>("previousMonitorInfo", MonitorInfo.class);
                        previousInfoState = getRuntimeContext().getState(stateDescriptor);
                    }

                    @Override
                    public void flatMap(MonitorInfo currentInfo, Collector<Violation> out) throws Exception {
                        // 获取前一条记录
                        MonitorInfo previousInfo = previousInfoState.value();

                        // 更新状态为当前记录
                        previousInfoState.update(currentInfo);

                        // 如果前一条记录存在
                        if (previousInfo != null) {
                            // 计算时间差（秒）
                            long timeDiff = Math.abs(currentInfo.getActionTime() - previousInfo.getActionTime());

                            // 检测套牌条件：
                            // 1. 两次记录时间差小于10秒
                            // 2. 不同监控点
                            // 3. 相同车牌号
                            if (timeDiff < 10 &&
                                    !currentInfo.getMonitorId().equals(previousInfo.getMonitorId())) {

                                // 生成违规记录
                                Violation violation = new Violation(
                                        0,
                                        currentInfo.getCar(),
                                        "涉嫌套牌",
                                        System.currentTimeMillis()
                                );

                                // 输出到下游
                                out.collect(violation);
                                System.out.println("检测到套牌车辆: " + violation);
                            }
                        }
                    }
                });

        resultDs.print("检测结果: ");


        // 在flatMap之后添加去重处理
        // ttl 处理
        SingleOutputStreamOperator<Violation> deduplicatedDs = resultDs
                .keyBy(Violation::getCar)  // 按车牌号分组
                .process(new KeyedProcessFunction<String, Violation, Violation>() {

                    private transient ValueState<Boolean> processedState;  // 状态：标记是否已处理过该车牌号

                    @Override
                    public void open(Configuration parameters) {
                        ValueStateDescriptor<Boolean> descriptor = new ValueStateDescriptor<>(
                                "deduplication-state",
                                Boolean.class
                        );
                        // 配置状态TTL（例如：1小时，避免状态无限增长）
                        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.minutes(1))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                                .build();
                        descriptor.enableTimeToLive(ttlConfig);
                        processedState = getRuntimeContext().getState(descriptor);
                    }

                    @Override
                    public void processElement(Violation violation, Context ctx, Collector<Violation> out) throws Exception {
                        String car = violation.getCar();
                        Boolean existed = processedState.value();

                        if (existed == null) {  // 首次处理该车牌号的违规记录
                            processedState.update(true);  // 标记为已处理
                            out.collect(violation);  // 输出到下游
                        } else {
                            // 重复记录，不输出（可添加日志监控）
                            System.out.println("重复记录：" + car );
                        }
                    }
                });


        // 将去重后的数据流写入MySQL
        // 4. 数据输出到MySQL
        deduplicatedDs.addSink(JdbcSink.sink(
                "INSERT INTO t_violation_list (car, violation, create_time) VALUES (?, ?, ?) "// +
                      //  "ON DUPLICATE KEY UPDATE violation=VALUES(violation), create_time=VALUES(create_time)",
                ,new JdbcStatementBuilder<Violation>() {
                    @Override
                    public void accept(PreparedStatement ps, Violation violation) throws SQLException {
                        ps.setString(1, violation.getCar());
                        ps.setString(2, violation.getViolation());
                        ps.setLong(3, violation.getCreateTime());
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(1)
                        .withBatchIntervalMs(1000)
                        .withMaxRetries(3)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUrl("jdbc:mysql://node101:3306/flink_project?useSSL=false&serverTimezone=Asia/Shanghai")
                        .withUsername("root")
                        .withPassword("123456")
                        .build()
        ));

        // 5. 执行任务
        env.execute("TaoPaiCarDetectionJob");
    }
}
